# -*- coding: utf-8 -*-
# @Time    : 2024/10/14 22:15
# @Author  : cx
# @File    : websockets_server.py
# @Software: PyCharm
import asyncio  # 协程异步
import csv
import json
import uuid
from datetime import datetime
from typing import Dict  # 类型指定

import websockets

# 异步读取文件
# 加载MYSQL操作
from mysql_database_utils_Y import AsyncMYSQL_L

# 定义全局变量 db
dbc = None
# 心跳检测间隔（秒）
HEARTBEAT_INTERVAL = 60


async def uuid_cz(yhid):
    """

    :param yhid:
    :return:
    如果CSV文件中存在UUID值为 'cx04' 的记录，函数将返回 True；如果不存在，函数将返回 False
    """
    try:
        with open('./记录.csv', 'r', newline='', encoding='utf-8') as f:
            reader = csv.reader(f)
            headers = next(reader)  # 跳过表头
            # 检查是否有与yhid相同的uuid
            for row in reader:
                if row[headers.index('uuid')] == yhid:
                    return True
            return False
    except FileNotFoundError:
        print('文件未找到，请检查文件路径是否正确。')
        return False
    except Exception as e:
        print(f'读取CSV文件时发生错误：{e}')
        return False


# 存储连接的映射表，键是UUID，值是WebSocket连接对象
connections: Dict[str, websockets.WebSocketServerProtocol] = {}
# 心跳检测
connections_zt: Dict[str, bool] = {}


# asyncio 的定时任务
async def heartbeat_check():
    """定期向所有客户端发送心跳并检查响应"""
    while True:
        await asyncio.sleep(HEARTBEAT_INTERVAL)  # 延时,定时
        # 遍历键的副本，原地修改字典
        # list(connections_zt.keys())把字典 connections_zt 的所有键取出来放在列表key里
        for key in list(connections_zt.keys()):
            if connections_zt[key]:  # 值为True时
                connections_zt[key] = False
            else:  # 值为False时
                await connections[key].close()  # 断开连接
                del connections[key]  # 删除映射表键值
                del connections_zt[key]  # 删除心跳表键值


async def close_websocket(websocket):
    await websocket.close()  # 关闭连接


# 将客户端的用户 token 和 WebSocket 连接绑定，并记录到服务端的全局连接池中
async def register(user_uuid, websocket):
    """
    从映射表中创建对应的 WebSocket 连接记录和状态记录,并把状态记录标记为True
    :param user_uuid: 握手令牌
    :param websocket: websocket连接对象
    :return:
    """
    # 检查 websocket 是否已存在于 connections 中
    # connection_id = next((id for id, ws in connections.items() if ws is websocket), None)
    # if connection_id:
    # 如果找到匹配的 websocket，删除原记录
    # del connections[connection_id]
    # 添加或更新连接
    # 生成一个唯一的UUID
    # user_uuid = str(uuid.uuid4())
    connections[user_uuid] = websocket
    connections_zt[user_uuid] = True
    # print(f"客户端 {user_uuid} 已连接。")
    # print(f"客户端 {connections[user_uuid]} 已连接。")


async def unregister(websocket):
    """
    从映射表中移除对应的 WebSocket 连接记录和状态记录
    :param websocket:
    :return:
    """
    connection_id = next((id for id, ws in connections.items() if ws is websocket), None)
    if connection_id:
        del connections[connection_id]  # 删除websocket连接ID信息
        del connections_zt[connection_id]  # 删除websocket连接ID心跳信息
        print(f"客户端 {connection_id} 已断开连接。")
        """ # 在连接数量达到或超过 1 个时，遍历并打印所有客户端的 ID 以及对应的 WebSocket 对象信息
            # connections.items()会返回由键值对组成的迭代器
        if len(connections) >= 1:
            for client_id, websocket in connections.items():
                print(f"客户端ID: {client_id}, WebSocket对象: {websocket}")
        else:
            print("wu l")
        """


async def send_to_client(connection_id, message):
    # 通过 ID 发送消息给特定的客户端
    if connection_id in connections:
        await connections[connection_id].send(message)  # 通过 ID 发送消息给特定的客户端
        return True
    else:
        print(f"无法发送消息，客户端 {connection_id} 不存在。")
        return False


# 消息分类处理
async def kong_zhi_ff(sj, db):
    sql = "SELECT device_token FROM devices WHERE location = %s and device_name = %s"
    result = await db.execute_query(sql, sj["ma"])
    match sj.get("location"):
        case 1:
            pass
        case 2:
            pass
        case 3:
            pass
        case 4:
            pass
        case 5:
            pass
        case 6:
            pass
        case 7:
            pass
        case _:
            # 当gong_neng的值不是"gl"时执行的代码
            pass


async def yan_zheng(response):
    """验证响应是否成功（同时检查 status 和 code）"""
    return (
            response is not None and
            response.get("status") == "success" and
            response.get("code") == 1
    )


async def xin_tiao(sj, db):
    sql = "SELECT device_token FROM devices WHERE device_token = %s"
    token = await db.execute_query(sql, sj["token"])
    if token:
        # 获取当前时间
        now = datetime.now()
        # 修改数据
        sql = "UPDATE devices SET last_online_time = %s WHERE device_token = %s"
        sql_cs = (now, sj["sx_token"])
        pd = await db.execute_operate(sql, sql_cs)
        if pd:
            return {
                "status": "success",
                "code": 1
            }
        else:
            return {
                "status": "fail",
                "code": 109,
                "message": "保存更改终端最后在线时间失败"
            }
    else:
        return {
            "status": "fail",
            "code": 106,
            "message": "数据库中无握手令牌相关信息 "
        }


async def shang_bao(sj, db):
    sql = "SELECT created_token FROM device_ma WHERE device_token = %s"
    result = await db.execute_query(sql, sj["token"])
    if result:
        # 获取当前时间
        now = datetime.now()
        days_integer = now - result[0][0]
        days_xiao_shi = days_integer.total_seconds() // 3600  # 总小时数（整数）
        # 令牌发放时间大于2天,就通知换令牌
        if 48 <= days_xiao_shi < 240:
            return {
                "status": "fail",
                "code": 108,
                "message": " 更换令牌 "
            }
        # 令牌发放时间大于10天,需要从新申请
        if days_xiao_shi >= 240:
            return {
                "status": "fail",
                "code": 107,
                "message": " 令牌失效,从新申请 "
            }
        # 处理上报
        match sj["lei_xing"]:
            case 1:
                pass
            case 2:
                pass
            case _:
                pass
        return {
            "status": "success",
            "code": 1
        }
    else:
        return {
            "status": "fail",
            "code": 106,
            "message": "数据库中无握手令牌相关信息 "
        }


async def kong_zhi(sj, db):
    sql = "SELECT created_token FROM device_ma WHERE device_token = %s"
    result = await db.execute_query(sql, sj["token"])
    if result:
        # 获取当前时间
        #now = datetime.now()
        #days_integer = now - result[0][0]
        #days_xiao_shi = days_integer.total_seconds() // 3600  # 总小时数（整数）
        # 令牌发放时间大于2天,就通知换令牌

        # 处理
        sql = "SELECT device_token FROM devices WHERE location = %s and device_name = %s"
        sql_cs = (sj["location"], sj["device_name"])
        result = await db.execute_query(sql, sql_cs)
        if result:
            return {
                "status": "success",
                "code": 1,
                "data": result[0][0]  # 指令下发的终端对象的 token令牌
            }
    else:
        return {
            "status": "fail",
            "code": 106,
            "message": "数据库中无握手令牌相关信息 "
        }


async def zhu_ce(sj, db):
    # {\"qing_qiu\": 1,\"code\": \"9C9E6EEBB078\", \"ma\": \"93352674dfaa450f81980801d9e72ed4\"}
    ss = {"zd_code": "00:11:22:33:44:55", "zd_ma": "1234567890abcdef1234567890abcdef", "zd_id": 12345,
          "zd_password": "password123", "zd_location": 1}
    sss = "{\"zd_code\": \"00:11:22:33:44:55\", \"zd_ma\": \"1234567890abcdef1234567890abcdef\", \"zd_id\": 12345, \"zd_password\": \"password123\", \"zd_location\": 1}"
    shua_xin_token = uuid.uuid4().hex
    sql = "SELECT device_code FROM device_ma WHERE device_ma = %s"
    print("zhu_ce  01")
    result = await db.execute_query(sql, sj["ma"])
    if result:
        if sj["code"] == result[0][0]:
            sql = "SELECT device_code FROM devices WHERE device_code = %s"
            result = await db.execute_query(sql, sj["code"])
            if result is None:
                return {
                    "status": "fail",
                    "code": 105,
                    "message": "数据库中无此网卡地址code "
                }
            # 获取当前时间
            now = datetime.now()
            # uuid格式的令牌
            token = uuid.uuid4().hex
            # 插入一条数据
            sql = """
            INSERT INTO devices(device_code, device_type, device_id, device_password, device_name,device_shuo_ming,location, status, device_token, created_token,device_shua_xin_token,created_shua_xin_token)
                            VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)
            """
            sql_cs = (
                sj["code"], sj["type"], sj["id"], sj["password"], sj["device_name"], sj["device_shuo_ming"],
                sj["location"], sj["status"], token, now,
                shua_xin_token,
                now)
            pd = await db.execute_operate(sql, sql_cs)
            if pd:
                return {
                    "status": "success",
                    "code": 1,
                    "data": token,  # 实际业务数据
                    "shua_xin_token": shua_xin_token
                }
            else:
                return {
                    "status": "fail",
                    "code": 104,
                    "message": "注册时保存数据到数据库失败"
                }


async def huo_qu_token(sj, db, websocket):
    sql = "SELECT created_shua_xin_token FROM devices WHERE device_shua_xin_token = %s"
    result = await db.execute_query(sql, sj["sx_token"])
    if result:
        # 获取当前时间
        now = datetime.now()
        days_integer = now - result[0][0]
        days_tian = days_integer.days  # 整数天数（忽略时间部分）
        if days_tian <= 180:
            # 修改数据
            sql = "UPDATE devices SET device_token = %s, created_token = %s WHERE device_shua_xin_token = %s"
            token = uuid.uuid4().hex
            sql_cs = (token, now, sj["sx_token"])
            pd = await db.execute_operate(sql, sql_cs)
            if pd:
                return {
                    "status": "success",
                    "code": 1,
                    "data": token  # 实际业务数据
                }
            else:
                return {
                    "status": "fail",
                    "code": 101,
                    "message": "保存令牌到数据库失败"
                }

        else:
            return {
                "status": "fail",
                "code": 102,
                "message": "刷新令牌过期"
            }  # 刷新令牌过期
    else:
        return {
            "status": "fail",
            "code": 103,
            "message": "无刷新令牌记录"
        }  # 无刷新令牌记录


async def connections_jian_ce(sj, websocket):
    pass


async def zd_yi_chang_xia_fa(sx_token, dai_ma, websocket):
    json_data = {
        "xia_fa": 4,
        "sx_token": sx_token,
        "yi_chang": dai_ma
    }
    xf = json.dumps(json_data)
    await websocket.send(xf)  # 发送消息给当前客户端


async def zd_kong_zhi(token, sj):
    """
    依据connections[token]发送指令到被控终端
    :param token: 需要接收指令的终端对应token
    :param sj: 发送指令终端(请求来源)
    :return:
    """
    json_data = {
        "xia_fa": 3,
        "qing_qiu_token": sj["token"],
        "token": token,
        "location": sj["location"],
        "device_name": sj["device_name"],
        "zhi_ling": sj["zhi_ling"]
    }
    xf = json.dumps(json_data)
    # 给指定连接推送消息
    if await send_to_client(token, xf):
        pass


# 异步函数处理客户端连接,websocket: WebSocket 连接对象，用于收发消息。path: 客户端连接的 URL 路径（未在代码中使用，但框架要求保留）
async def handle_client(websocket: websockets.WebSocketServerProtocol, path: str, db):
    try:
        # 异步迭代接收客户端发送的消息，直到连接关闭
        async for message in websocket:
            try:
                sj = json.loads(message)
            except json.JSONDecodeError as e:
                await websocket.close(4101, "非法")  # 关闭连接
                print(f"json错误,不是json数据: {e}")
            else:
                print(sj)
                print(f"客户端地址: {websocket.remote_address}")
                print(f"客户HTTP请求头: {websocket.request_headers}")
                print(f"客户端HTTP响应头: {websocket.response_headers}")
                print(f"请求路径: {path}")

                """
                # 使用 get() 方法，不存在时返回 None
                # zd_token = sj.get("token")  # 不存在时返回 None
                # zd_token = sj.get("token", "default_token")  # 不存在时返回 "default_token"
                
                zd_code = sj["code"] # 网卡地址
                zd_id = sj["id"]# id
                zd_password = sj["password"]# 密码
                zd_type = sj["type"] # 类型
                zd_location = sj["location"] # 位置
                zd_qing_qiu = sj["qing_qiu"] # 请求类型
                zd_status = sj["status"] #终端当前状态(泛指开关类)
                zd_token = sj["token"] #令牌
                zd_ma =sj["ma"]# 32位设备码
                """

                match sj["qing_qiu"]:
                    case 1:
                        fh = await xin_tiao(sj, db)
                        if await yan_zheng(fh):
                            if sj["token"] not in connections:
                                await register(sj["yhid"], websocket)  # 将客户端的用户 ID 和 WebSocket 连接绑定
                                print(f'{sj["token"]} 登陆成功')
                            else:
                                connections_zt[sj["token"]] = True  # 更新心跳记录
                            json_data = {
                                "xia_fa": 1,
                                "token": sj["token"]
                            }
                            xf = json.dumps(json_data)
                            connections_zt[sj["token"]] = True  # 更新心跳记录
                            await websocket.send(xf)  # 发送消息给当前客户端
                        else:
                            await websocket.close(4100, "无连接权限")  # 关闭连接

                    case 2:  # 上报
                        print("qing_qiu   - 1")
                        fh = await shang_bao(sj, db)
                        if await yan_zheng(fh):
                            if sj["token"] not in connections:
                                await register(sj["yhid"], websocket)  # 将客户端的用户 ID 和 WebSocket 连接绑定
                                print(f'{sj["token"]} 登陆成功')
                            json_data = {
                                "xia_fa": 3,
                                "sx_token": sj["sx_token"],
                                "token": sj["token"],
                                "zhuang_tai": 1
                            }
                            xf = json.dumps(json_data)
                            connections_zt[sj["token"]] = True  # 更新心跳记录
                            await websocket.send(xf)  # 发送消息给当前客户端
                        else:
                            if fh.get("code") == 107:  # 握手令牌失效,从新申请
                                await zd_yi_chang_xia_fa(sj["sx_token"], 107, websocket)
                            if fh.get("code") == 108:  # 更换握手令牌
                                await zd_yi_chang_xia_fa(sj["sx_token"], 108, websocket)
                    case 3:  # 控制
                        print("qing_qiu   - 2")
                        fh = await kong_zhi(sj, db)
                        if await yan_zheng(fh):
                            connections_zt[sj["token"]] = True  # 更新心跳记录
                            await zd_kong_zhi(fh.get("data"), sj)
                    case 4:  # 获取令牌
                        zd_token = await huo_qu_token(sj, db, websocket)
                        if await yan_zheng(zd_token):
                            json_data = {
                                "xia_fa": 4,
                                "sx_token": sj["sx_token"],
                                "token": zd_token.get("data")
                            }
                            xf = json.dumps(json_data)
                            await websocket.send(xf)  # 发送消息给当前客户端
                            connections_zt[sj["token"]] = True  # 更新心跳记录
                        else:
                            if zd_token.get("code") == 102:
                                await zd_yi_chang_xia_fa(sj["sx_token"], 102, websocket)
                    case 5:
                        pass

                    case 6:  # 注册
                        pd = await zhu_ce(sj, db)
                        if await yan_zheng(pd):
                            json_data = {
                                "xia_fa": 6,
                                "sx_token": pd.get("sx_token"),
                                "token": pd.get("data")
                            }
                            xf = json.dumps(json_data)
                            await register(pd.get("data"), websocket)  # 将客户端的用户 token 和 WebSocket 连接绑定
                            await websocket.send(xf)  # 发送消息给当前客户端
                    case 7:
                        pass
                    case _:
                        # 当以上条件都不满足时执行的代码
                        await close_websocket(websocket)  # 断开当前连接
                        pass
                """
                # 是否登陆,如果connections里没有sj["yhid"]键
                if sj["yhid"] not in connections:
                    print(message)
                    if uuid_cz(sj["yhid"]):  # 检测用户
                        await register(sj["yhid"], websocket)  #将客户端的用户 ID 和 WebSocket 连接绑定
                        print(f'{sj["yhid"]} 登陆成功')
                    else:
                        await websocket.close(4100, "无此用户")  #关闭连接
                else:  # 已经登陆,需要单独的指令
                    # 这里可以添加服务器端的逻辑来处理消息
                    print(f"收到来自客户端 {sj['yhid']} 的消息: {message}")
                    #消息分类处理:
                    await message_cl(sj["yhid"], sj)
                    #给全部连接推送消息
                    send_tasks = [ws.send(f"服务器消息: {sj['yhid']} 连接") for ws in connections.values()]  #给所有客户端发送信息
                    # 为每个连接创建发送任务，通过 asyncio.gather 并发执行
                    #await asyncio.gather(*[asyncio.create_task(task) for task in send_tasks])
                    await asyncio.gather(*send_tasks, return_exceptions=True)  # 自动处理协程到Task的转换,使用 return_exceptions=True 参数让 gather 返回异常而非抛出
                    # 给指定连接推送消息
                    await send_to_client(sj["yhid"], f'niubi {sj["yhid"]}')
"""

    except websockets.exceptions.ConnectionClosed as e:
        # 如果客户端断开连接，会捕获到这个异常
        print(f"客户端断开连接: {e}")
    except Exception as e:
        print(f"出错啦: {e}")
    finally:
        await unregister(websocket)


async def qi_dong():
    # global db # 声明对全局变量的修改
    db = AsyncMYSQL_L('192.168.1.88', 3306, 'a18222970', 'cx123100', 'zhi_neng_wang_guan')
    await db.create_pool()  # 建立MYSQL连接池
    # server = await websockets.serve(handle_client, "0.0.0.0", 8080) # 配置,启动服务
    # lambda 函数通常用于简单的函数定义; lambda函数调用handle_client函数,并传递了3个参数给他,lambda函数本身成了websockets.serve的回调函数
    server = await websockets.serve(lambda ws, p: handle_client(ws, p, db), "0.0.0.0", 8080)  # 配置,启动服务
    print("WebSocket服务器已启动，监听端口8080...")
    # 启动心跳检测任务
    heartbeat_task = asyncio.create_task(heartbeat_check())
    try:
        await server.wait_closed()  # 保持服务器运行，直到手动停止
    finally:
        heartbeat_task.cancel()  # 取消心跳任务
        await heartbeat_task  # 等待任务完全取消
        await db.close_pool()  # 释放数据库连接池


def run():
    # 配置WebSocket服务器的参数。handle_client：客户端连接时的回调函数，用于处理消息和连接逻辑。
    # "0.0.0.0"：服务器监听所有网络接口（允许外部设备连接）。8080：服务器监听的端口号
    # lambda 函数通常用于简单的函数定义; lambda函数调用handle_client函数,并传递了3个参数给他,lambda函数本身成了websockets.serve的回调函数
    start_server = websockets.serve(lambda ws, p: handle_client(ws, p, dbc), "0.0.0.0", 8080)
    """
    执行异步任务以启动服务器。
    asyncio.get_event_loop()：获取当前事件循环,
    run_until_complete(start_server)：阻塞当前线程，直到服务器完成启动（绑定端口并开始监听）
    """
    asyncio.get_event_loop().run_until_complete(start_server)
    # 使事件循环无限运行，持续处理客户端请求。若无此调用，程序会在启动服务器后立即退出。
    # run_forever()确保事件循环监听新连接、消息和异步事件
    asyncio.get_event_loop().run_forever()


if __name__ == '__main__':
    asyncio.run(qi_dong())

'''
/www/server/pyporject_evn/websockets_fw_venv/bin/python3 -u /www/wwwroot/websockets_fw/websockets_server.py
'''
